Platform-Triggered Rotation

The rotation itself is identical to device-triggered rotation. The device connects with the bootstrap certificate, runs the Phase 2 Fleet Provisioning flow, and reconnects with the new certificate. The difference is the trigger and the wrapper: instead of detecting expiry locally, the device discovers the rotation request as an AWS IoT Job and must report the outcome (SUCCEEDED or FAILED) back via the Jobs API.

This approach handles offline devices correctly. Unlike a direct MQTT push, a Job persists until the device comes online and processes it.

How the Cloud Side Works

The platform may create a rotation job for several reasons: certificate expiry (the most common), a CA rotation, a security incident, or any other reason the platform team determines. From the firmware perspective, all of these arrive as an identical rotate-certificate job. The device does not need to know why rotation was requested.

For expiry-driven rotation specifically:

1. AWS IoT Defender's DEVICE_CERTIFICATE_EXPIRING_CHECK audit runs on a schedule and flags certificates expiring within 30 days
2. Defender publishes audit findings to SNS
3. A Lambda function processes the findings and identifies the affected Things
4. Lambda calls CreateJob targeting each affected Thing by ARN
5. The Job persists until the device processes it

The device does not need to do anything special to be targeted. The platform handles job creation automatically.

Job Document Format

{
    "operation": "rotate-certificate"
}

All CPP job types use the same operation field for dispatch. The device must inspect this field before claiming a job.

MQTT Topics

Topic	Direction	Purpose
$aws/things/{MPBID}/jobs/notify-next	Subscribe	Notified when a new pending job becomes available
$aws/things/{MPBID}/jobs/get	Publish	Get list of all pending job IDs (without starting any)
$aws/things/{MPBID}/jobs/get/accepted	Subscribe	Receive pending job list
$aws/things/{MPBID}/jobs/get/rejected	Subscribe	Error if request malformed
$aws/things/{MPBID}/jobs/{jobId}/get	Publish	Describe a specific job and retrieve its document
$aws/things/{MPBID}/jobs/{jobId}/get/accepted	Subscribe	Receive job document
$aws/things/{MPBID}/jobs/{jobId}/get/rejected	Subscribe	Error if job not found
$aws/things/{MPBID}/jobs/{jobId}/update	Publish	Report job status (IN_PROGRESS, SUCCEEDED, FAILED)
$aws/things/{MPBID}/jobs/{jobId}/update/accepted	Subscribe	Confirmation of status update
$aws/things/{MPBID}/jobs/{jobId}/update/rejected	Subscribe	Error if status update rejected

Handling Multiple Job Types

The device may have jobs of different types queued simultaneously (e.g. an OTA firmware update and a certificate rotation). The key constraint is that claiming a job commits the device to completing it — the Jobs service marks it IN_PROGRESS and expects a terminal status (SUCCEEDED or FAILED) within the step timeout.

For this reason the device should not use StartNextPendingJobExecution, which blindly claims whichever job is at the front of the queue. Instead:

1. Use GetPendingJobExecutions to retrieve all queued job IDs without starting any of them
2. Use DescribeJobExecution on each to inspect the job document and find the rotation job
3. Only then claim it by publishing IN_PROGRESS via UpdateJobExecution

This way the rotation logic never accidentally claims an OTA job, and vice versa. Each job type handler is responsible for finding and claiming its own jobs.

Once a job is marked IN_PROGRESS, the Jobs service expects a terminal status (SUCCEEDED or FAILED) within the configured step timeout. Contact the CPP team for the specific timeout value configured for rotation jobs. If the device does not report a terminal status in time, the Jobs service will mark the execution TIMED_OUT. The platform monitors for this and will escalate persistent failures.

Rotation Sequence

Step-by-Step Implementation

On every connect (and whenever notify-next fires), the device should scan for a pending rotation job.

1. Get all pending job IDs

Publish to $aws/things/{MPBID}/jobs/get:

{}

The accepted response contains queuedJobs and inProgressJobs arrays, each holding JobExecutionSummary objects with jobId and executionNumber. The job document is not included at this stage. The executionNumber is a monotonically increasing counter used for optimistic concurrency control — echo it back in every UpdateJobExecution call, otherwise the update will be rejected.

A job appearing in inProgressJobs means the device previously claimed it and crashed or lost power before reporting a terminal status. The jobId and executionNumber are available directly from this response — no NVM needed. If the new certificate is already in secure storage and connects successfully, skip straight to reporting SUCCEEDED. Otherwise re-attempt the full rotation and re-send IN_PROGRESS before disconnecting.

2. Inspect each job to find the rotation job

For each jobId in the pending list, publish to $aws/things/{MPBID}/jobs/{jobId}/get:

{ "includeJobDocument": true }

Check the execution.jobDocument.operation field in the response. Skip jobs whose operation is not rotate-certificate and handle them with their own dispatch logic.

3. Claim the rotation job

Publish to $aws/things/{MPBID}/jobs/{jobId}/update:

{
    "status": "IN_PROGRESS",
    "statusDetails": { "step": "starting-rotation" },
    "executionNumber": <executionNumber>
}

4. Perform the rotation

Disconnect the operational session and follow the device-triggered rotation steps using the bootstrap certificate.

5. Report SUCCEEDED or FAILED

After reconnecting with the new operational certificate, publish the final job status:

{
    "status": "SUCCEEDED",
    "statusDetails": { "newCertificateId": "<64-char-hex-id>" },
    "executionNumber": <executionNumber>
}

If rotation fails at any step, reconnect with the old operational certificate (if still valid) and report failure:

{
    "status": "FAILED",
    "statusDetails": { "reason": "fleet-provisioning-timeout" },
    "executionNumber": <executionNumber>
}

The platform monitors failed job executions. Persistent failures will escalate to the JEDI DIoTS team.

Key and CSR Requirements

Generate an RSA-2048 key pair. The provisioning template validates the CSR subject. The following fields are required and must match exactly:

Field	Value
CN (Common Name)	Device type string (e.g. bridge)
GN (Given Name)	Device MPBID (e.g. FFFF000001)
O (Organization)	Milwaukee Tool
OU (Organizational Unit)	Connected Products
C (Country)	US
ST (State)	WI
L (Locality)	Brookfield

The MPBID in the GN field is how the provisioning template associates the new certificate with the correct Thing. An incorrect or missing GN will cause RegisterThing to fail.

Reference Implementation

The following script shows the complete platform-triggered rotation flow. It assumes the device is already connected with its operational certificate and has picked up a rotate-certificate job. Adapt the MQTT calls to your platform's client library.

import json
import time
from awscrt import mqtt
from awsiot import iotidentity, mqtt_connection_builder
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import hashes, serialization
from cryptography import x509
from cryptography.x509.oid import NameOID

# --- Configuration ---
DEVICE_MPBID          = "FFFF000001"
MQTT_ENDPOINT         = "mqtt.prod.iot.digital.milwaukeetool.com"
OPERATIONAL_CERT_PEM  = "..."   # Current operational cert PEM
OPERATIONAL_KEY_PEM   = "..."   # Current operational key PEM
BOOTSTRAP_CERT_PEM    = "..."   # Bootstrap cert PEM (permanent)
BOOTSTRAP_KEY_PEM     = "..."   # Bootstrap key PEM (permanent)
DEVICE_TYPE           = "bridge"

PROVISIONING_TEMPLATE = "generic-provisioning-template"
RESPONSE_TIMEOUT_SECS = 15   # increase for cellular or high-latency links (60+ seconds recommended)


def build_connection(cert_pem: str, key_pem: str) -> mqtt.Connection:
    return mqtt_connection_builder.mtls_from_bytes(
        endpoint=MQTT_ENDPOINT,
        cert_bytes=cert_pem.encode(),
        pri_key_bytes=key_pem.encode(),
        client_id=DEVICE_MPBID,
        clean_session=False,   # always resubscribe explicitly after a cert switch; do not rely on the
                               # broker restoring persistent session subscriptions across connections
                               # made with different certificates
        keep_alive_secs=30,
    )


def wait_for(response: dict, operation: str):
    for _ in range(RESPONSE_TIMEOUT_SECS):
        if response["error"]:
            raise response["error"]
        if response["data"]:
            return response["data"]
        time.sleep(1)
    raise TimeoutError(f"Timed out waiting for {operation}")


def update_job(conn: mqtt.Connection, job_id: str, execution_number: int,
               status: str, status_details: dict):
    conn.publish(
        topic=f"$aws/things/{DEVICE_MPBID}/jobs/{job_id}/update",
        payload=json.dumps({
            "status": status,
            "statusDetails": status_details,
            "executionNumber": execution_number,
        }).encode(),
        qos=mqtt.QoS.AT_LEAST_ONCE,
    )[0].result()


# --- Step 1: Connect with operational cert and scan for a pending rotation job ---
# We use GetPendingJobExecutions + DescribeJobExecution rather than StartNext so
# we inspect job documents before claiming anything. This avoids accidentally
# claiming an OTA or other job type that this code doesn't know how to complete.
op_conn = build_connection(OPERATIONAL_CERT_PEM, OPERATIONAL_KEY_PEM)
op_conn.connect().result()

pending_response = {"data": None, "error": None}

op_conn.subscribe(
    topic=f"$aws/things/{DEVICE_MPBID}/jobs/get/accepted",
    qos=mqtt.QoS.AT_LEAST_ONCE,
    callback=lambda topic, payload, **kw: pending_response.update({"data": json.loads(payload)}),
)[0].result()

op_conn.subscribe(
    topic=f"$aws/things/{DEVICE_MPBID}/jobs/get/rejected",
    qos=mqtt.QoS.AT_LEAST_ONCE,
    callback=lambda topic, payload, **kw: pending_response.update(
        {"error": Exception(f"jobs/get rejected: {payload}")}
    ),
)[0].result()

op_conn.publish(
    topic=f"$aws/things/{DEVICE_MPBID}/jobs/get",
    payload=json.dumps({}).encode(),
    qos=mqtt.QoS.AT_LEAST_ONCE,
)[0].result()

pending = wait_for(pending_response, "GetPendingJobExecutions")
all_jobs = pending.get("queuedJobs", []) + pending.get("inProgressJobs", [])

if not all_jobs:
    op_conn.disconnect().result()
    raise SystemExit("No pending jobs")

# Scan each job to find a rotate-certificate job without claiming any of them
rotation_job_id      = None
rotation_exec_number = None

for summary in all_jobs:
    job_id      = summary["jobId"]
    exec_number = summary["executionNumber"]

    describe_response = {"data": None, "error": None}

    op_conn.subscribe(
        topic=f"$aws/things/{DEVICE_MPBID}/jobs/{job_id}/get/accepted",
        qos=mqtt.QoS.AT_LEAST_ONCE,
        callback=lambda topic, payload, **kw: describe_response.update({"data": json.loads(payload)}),
    )[0].result()

    op_conn.subscribe(
        topic=f"$aws/things/{DEVICE_MPBID}/jobs/{job_id}/get/rejected",
        qos=mqtt.QoS.AT_LEAST_ONCE,
        callback=lambda topic, payload, **kw, _jid=job_id: describe_response.update(
            {"error": Exception(f"jobs/{_jid}/get rejected: {payload}")}
        ),
    )[0].result()

    op_conn.publish(
        topic=f"$aws/things/{DEVICE_MPBID}/jobs/{job_id}/get",
        payload=json.dumps({"includeJobDocument": True}).encode(),
        qos=mqtt.QoS.AT_LEAST_ONCE,
    )[0].result()

    described = wait_for(describe_response, f"DescribeJobExecution({job_id})")
    operation = described.get("execution", {}).get("jobDocument", {}).get("operation")

    if operation == "rotate-certificate":
        rotation_job_id      = job_id
        rotation_exec_number = exec_number
        break
    # Non-rotation jobs are left untouched. Their own handlers will claim them.

if rotation_job_id is None:
    op_conn.disconnect().result()
    raise SystemExit("No rotation job found in pending queue")

# --- Step 2: Claim the rotation job and mark IN_PROGRESS before disconnecting ---
update_job(op_conn, rotation_job_id, rotation_exec_number, "IN_PROGRESS", {"step": "starting-rotation"})
job_id      = rotation_job_id
exec_number = rotation_exec_number
op_conn.disconnect().result()

# --- Step 3: Connect with bootstrap cert and run Phase 2 Fleet Provisioning ---

bs_conn = build_connection(BOOTSTRAP_CERT_PEM, BOOTSTRAP_KEY_PEM)
bs_conn.connect().result()

identity_client = iotidentity.IotIdentityClient(bs_conn)

create_cert_response    = {"data": None, "error": None}
register_thing_response = {"data": None, "error": None}

identity_client.subscribe_to_create_certificate_from_csr_accepted(
    request=iotidentity.CreateCertificateFromCsrSubscriptionRequest(),
    qos=mqtt.QoS.AT_LEAST_ONCE,
    callback=lambda r: create_cert_response.update({"data": r}),
)[0].result()

identity_client.subscribe_to_create_certificate_from_csr_rejected(
    request=iotidentity.CreateCertificateFromCsrSubscriptionRequest(),
    qos=mqtt.QoS.AT_LEAST_ONCE,
    callback=lambda e: create_cert_response.update(
        {"error": Exception(f"CreateCertificateFromCSR rejected: {e.error_code}")}
    ),
)[0].result()

identity_client.subscribe_to_register_thing_accepted(
    request=iotidentity.RegisterThingSubscriptionRequest(template_name=PROVISIONING_TEMPLATE),
    qos=mqtt.QoS.AT_LEAST_ONCE,
    callback=lambda r: register_thing_response.update({"data": r}),
)[0].result()

identity_client.subscribe_to_register_thing_rejected(
    request=iotidentity.RegisterThingSubscriptionRequest(template_name=PROVISIONING_TEMPLATE),
    qos=mqtt.QoS.AT_LEAST_ONCE,
    callback=lambda e: register_thing_response.update(
        {"error": Exception(f"RegisterThing rejected: {e.error_code}")}
    ),
)[0].result()

new_private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)  # RSA-2048 required
new_private_key_pem = new_private_key.private_bytes(
    encoding=serialization.Encoding.PEM,
    format=serialization.PrivateFormat.TraditionalOpenSSL,
    encryption_algorithm=serialization.NoEncryption(),
).decode()

csr_pem = x509.CertificateSigningRequestBuilder().subject_name(x509.Name([
    x509.NameAttribute(NameOID.COMMON_NAME,              DEVICE_TYPE),
    x509.NameAttribute(NameOID.GIVEN_NAME,               DEVICE_MPBID),
    x509.NameAttribute(NameOID.ORGANIZATION_NAME,        "Milwaukee Tool"),
    x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, "Connected Products"),
    x509.NameAttribute(NameOID.COUNTRY_NAME,             "US"),
    x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME,   "WI"),
    x509.NameAttribute(NameOID.LOCALITY_NAME,            "Brookfield"),
])).sign(new_private_key, hashes.SHA256()).public_bytes(serialization.Encoding.PEM).decode()

identity_client.publish_create_certificate_from_csr(
    request=iotidentity.CreateCertificateFromCsrRequest(certificate_signing_request=csr_pem),
    qos=mqtt.QoS.AT_LEAST_ONCE,
).result()

cert_data = wait_for(create_cert_response, "CreateCertificateFromCSR")
new_cert_pem    = cert_data.certificate_pem
new_cert_id     = cert_data.certificate_id
ownership_token = cert_data.certificate_ownership_token

identity_client.publish_register_thing(
    request=iotidentity.RegisterThingRequest(
        template_name=PROVISIONING_TEMPLATE,
        certificate_ownership_token=ownership_token,
        parameters={"MPBID": DEVICE_MPBID},
    ),
    qos=mqtt.QoS.AT_LEAST_ONCE,
).result()

wait_for(register_thing_response, "RegisterThing")
bs_conn.disconnect().result()

# Store new_cert_pem and new_private_key_pem to secure storage on the device.
# Never log the private key.

# --- Step 4: Reconnect with new operational cert and report SUCCEEDED ---

new_op_conn = build_connection(new_cert_pem, new_private_key_pem)
new_op_conn.connect().result()

update_job(new_op_conn, job_id, exec_number, "SUCCEEDED", {"newCertificateId": new_cert_id})

print(f"Rotation complete. New certificate: {new_cert_id}")

Failure and Recovery

If rotation fails after claiming IN_PROGRESS: Reconnect using the old operational certificate (if still valid) and report FAILED. The platform monitors failed executions and will re-create the job if needed.

If the old certificate has already expired: Connect directly with the bootstrap certificate, re-attempt the Phase 2 provisioning flow to obtain a new certificate, then reconnect with the new operational certificate and report SUCCEEDED.

Crash before reporting SUCCEEDED: On reboot, call GetPendingJobExecutions — the job will appear in inProgressJobs with the jobId and executionNumber in the response. If the new certificate is already in secure storage and connects successfully, report SUCCEEDED. Otherwise re-attempt the full rotation and re-send IN_PROGRESS before disconnecting. Worst case the step timeout fires and the platform re-creates the job.

Do not retry indefinitely within a single session. One attempt per boot is sufficient.

AWS Documentation

• AWS IoT Jobs: device workflow
• Jobs MQTT API reference
• IoT Device Defender certificate expiry check